package com.shujia.streaming

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo05StructureStreaming {
  def main(args: Array[String]): Unit = {

    // 构建SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo05StructureStreaming")
      .master("local[2]")
      .getOrCreate()

    // 不需要指定批次的大小 默认为100ms一个批次 默认会对状态进行维护
    val lineDFOnStream: DataFrame = spark
      .readStream
      .format("socket")
      .option("host", "master")
      .option("port", 8888)
      .load()

    // 完成wordCount

    // SQL的方式
    lineDFOnStream.createOrReplaceTempView("words_tb")

    val wordCntDF: DataFrame = spark.sql(
      """
        |select  t1.word
        |        ,count(*) as cnt
        |from (
        |    select  explode(split(value,",")) as word
        |    from words_tb
        |) t1 group by t1.word
        |""".stripMargin)

    // 如果数据是一个流 则将结果输出的时候不能直接使用show
    wordCntDF
      .writeStream
      .outputMode("complete")
      .format("console") // 将结果源源不断输出到控制台
    //      .start() // 启动任务
    //      .awaitTermination() // 等待停止

    // DSL处理
    import spark.implicits._
    import org.apache.spark.sql.functions._

    lineDFOnStream
      .select(explode(split($"value", ",")) as "word")
      .groupBy($"word")
      .agg(count("*") as "cnt")
      .writeStream

      /**
       * OutputMode指定流输出的方式：
       * append 追加 ： 只适用于一些简单的查询，不能应用在有聚合操作的流中
       * complete 完整 ：只适用于聚合操作之后的输出，会将所有的结果都输出
       * update 更新 ：只会将有变化的行的数据输出，不管有无聚合操作都可以使用，如果流中没有聚合操作则等价于Append模式
       */
      .outputMode("update")
      .format("console") // 将结果源源不断输出到控制台
      // 将每个批次的数据转换成DataFrame进行处理
      //      .foreachBatch((df,l)=>{
      //        df.write
      //          .format("jdbc")
      //          .option("url", "jdbc:mysql://master:3306/bukong?useSSL=false")
      //          .option("dbtable", "word_cnt")
      //          .option("user", "root")
      //          .option("password", "123456")
      //          .mode(SaveMode.Overwrite) // 会覆盖原有的表 并且表结构会发生变化
      //          .save()
      //      })
      .start() // 启动任务
      .awaitTermination() // 等待停止


  }

}
